package com.sgq.vue.redis.impl;

import com.alibaba.fastjson.JSON;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.sgq.vue.redis.CommonSerializer;
import com.sgq.vue.redis.RedisService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.types.Expiration;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import javax.annotation.Resource;
import java.lang.reflect.Type;
import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@Slf4j
public class RedisServiceImpl implements RedisService {

    @Value("${spring.redis.prefix}")
    private String prefix;

    public String getPrefix() {
        return prefix;
    }

    @Resource
    private RedisTemplate<String, byte[]> redisTemplate;

    private ObjectMapper objectMapper;

    @Autowired
    private CommonSerializer serializer;

    public RedisServiceImpl() {
        super();
        this.objectMapper = new ObjectMapper();
    }

    /**
     * 系列化：将给定的对象序列化为JSON字节数组
     *
     * @param value
     * @return 数组
     */
    public byte[] serialize(Object value) {
        byte[] bytes = null;
        try {
            bytes = objectMapper.writeValueAsBytes(value);
        } catch (Exception e) {
            log.error("Failed to serialize and save data to Redis", e);
        }
        return bytes;
    }

    /**
     * 系列化List集合
     *
     * @param values
     * @return
     */
    byte[][] serializeValues(Collection<?> values) {
        byte[][] rawValues = new byte[values.size()][];
        int i = 0;

        Object value;
        for (Iterator<?> iterator = values.iterator(); iterator.hasNext(); rawValues[i++] = this.serialize(value)) {
            value = iterator.next();
        }

        return rawValues;
    }

    /**
     * 反系列化：从Redis中检索字节数组并将其反序列化为给定类型的对象
     *
     * @param data      redis数据
     * @param valueType 转换对象
     * @return
     */
    public <V> V deserialize(byte[] data, Class<V> valueType) {
        if (data == null) {
            return null;
        }
        try {
            return (V) serializer.deserialize(data, valueType);
        } catch (Exception e) {
            log.error("Failed to retrieve and deserialize data from Redis", e);
        }
        return null;
    }

    /**
     * 系列化字符串
     *
     * @param value
     * @return
     */
    private byte[] serializeForString(String value) {
        return value == null ? null : value.getBytes(StandardCharsets.UTF_8);
    }

    /**
     * 获取指定变量中的value值
     *
     * @param key
     * @return
     * @Author: chenzumao
     */
    @Override
    public <V> List<V> hGetAllToList(String key, Class<V> clazz) {
        List<V> objectList = new ArrayList<>();
        byte[] keyBytes = serializeForString(key);
        if (this.hasKey(key)) {
            List<byte[]> dataBytes = this.redisTemplate.execute(connection -> connection.hVals(keyBytes), true);
            objectList = deserializeList(dataBytes, clazz);
        }

        return objectList;
    }

    /**
     * 根据key和多个field获取hash相应的value值 返回list
     *
     * @param key    主键key
     * @param fields 批量获取的副键
     * @param clazz  类
     * @return 集合
     * @Author: chenzumao
     */
    @Override
    public <V> List<V> hGetAllToList(String key, Collection<String> fields, Class<V> clazz) {
        List<V> objectList = new ArrayList<>();
        byte[] keyBytes = serializeForString(key);
        if (CollectionUtils.isEmpty(fields)) {
            return objectList;
        }
        List<String> collect = fields.stream().filter(item -> item != null && item != "").collect(Collectors.toList());
        if (CollectionUtils.isEmpty(collect)) {
            return objectList;
        }

        byte[][] rawHashKeys = new byte[collect.size()][];
        int counter = 0;
        for (String hashKey : fields) {
            rawHashKeys[counter++] = serializeForString(hashKey);
        }

        if (this.hasKey(keyBytes)) {
            List<byte[]> dataBytes = this.redisTemplate.execute(connection -> connection.hMGet(keyBytes, rawHashKeys), true);
            objectList = deserializeList(dataBytes, clazz);
        }

        return objectList;
    }


    /**
     * List反系列化
     *
     * @param dataBytes 查询到的数据
     * @param clazz     类
     * @return
     */
    private <V> List<V> deserializeList(List<byte[]> dataBytes, Class<V> clazz) {
        List<V> dataList = new ArrayList<>();
        if (CollectionUtils.isEmpty(dataBytes)) {
            return dataList;
        } else {
            Iterator<byte[]> iterator = dataBytes.iterator();
            while (iterator.hasNext()) {
                byte[] next = iterator.next();
                if (null == next || next.length == 0) {
                    continue;
                }
                V dataInfo = this.deserialize(next, clazz);
                if (null != dataInfo) {
                    dataList.add(dataInfo);
                }
            }
        }
        return dataList;
    }


    <V extends Collection<?>> V deserializeValues(Collection<byte[]> rawValues, Class<?> collectionType, Class<V> type) {
        if (rawValues == null) {
            return (V) new ArrayList<>();
        } else {
            Collection<Object> values = List.class.isAssignableFrom(collectionType) ? new ArrayList<>(rawValues.size()) : new LinkedHashSet<>(rawValues.size());
            Iterator<byte[]> var5 = rawValues.iterator();

            while (var5.hasNext()) {
                byte[] bs = var5.next();
                values.add(this.deserialize(bs, type));
            }

            return (V) values;
        }
    }

    /**
     * 把value系列化存byte
     *
     * @param value
     * @param <V>
     * @return
     */

    private <V> byte[] serializeToString(V value) {
        return null == value ? new byte[0] : this.serializer.serialize((Type) null, value);
    }

    /**
     * 基本类型数据操作
     *
     * @param key       key
     * @param value     value byte
     * @param expireSec 过期时间 毫秒
     */
    private void doPut(String key, final byte[] value, final long expireSec) {
        final byte[] keyBytes = this.serializeForString(key);
        this.redisTemplate.execute(new RedisCallback<Object>() {
            public String doInRedis(RedisConnection connection) throws DataAccessException {
                if (expireSec > 0L) {
                    connection.set(keyBytes, value, Expiration.milliseconds(expireSec), RedisStringCommands.SetOption.UPSERT);
                } else {
                    connection.set(keyBytes, value);
                }

                return null;
            }
        });
    }

    /**
     * 普通类型String 推送
     *
     * @param key  key
     * @param var2 value
     */
    @Override
    public <V> void put(String key, V var2, Long expireTime, final TimeUnit unit) {
        byte[] valueBytes = serializeToString(var2);
        long time = 0;
        if (null != expireTime) {
            time = unit.toMillis(expireTime);
        }
        this.doPut(key, valueBytes, time);
    }

    /**
     * 普通类型String 推送
     *
     * @param key  key
     * @param var2 value
     */
    @Override
    public <V> void put(String key, V var2) {
        byte[] valueBytes = serializeToString(var2);
        this.doPut(key, valueBytes, 0);
    }

    @Override
    public <V> V get(String key, Class<V> clazz) {
        byte[] byteValue = this.doGet(key);
        return byteValue == null ? null : deserialize(byteValue, clazz);
    }

    private byte[] doGet(final String key) {

        byte[] bytes = serializeForString(key);
        return this.redisTemplate.execute(new RedisCallback<byte[]>() {
            public byte[] doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.get(bytes);
            }
        });
    }

    /**
     * set数据读取一次读一个
     *
     * @param key
     * @param clazz
     * @return
     */
    @Override
    public <V> V setPop(String key, Class<V> clazz) {
        final byte[] keyBytes = this.serializeForString(key);
        byte[] hGetBytes = this.redisTemplate.execute(connection -> connection.sPop(keyBytes), true);
        return this.deserialize(hGetBytes, clazz);
    }

    /**
     * set数据添加（批量）
     *
     * @param key
     * @param values 对象集合
     * @return 影响条数
     */
    @Override
    public <V> Long setAddList(String key, List<V> values) {
        byte[][] listBytes = this.serializeValues(values);
        final byte[] keyBytes = this.serializeForString(key);
        return this.redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.sAdd(keyBytes, listBytes);
            }
        });
    }

    /**
     * set数据添加
     *
     * @param key   键
     * @param value 对象
     * @return 影响条数
     */
    @Override
    public <V> Long setAdd(String key, V value) {
        byte[] valueByte = this.serialize(value);
        final byte[] keyBytes = this.serializeForString(key);
        return this.redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.sAdd(keyBytes, valueByte);
            }
        });
    }

    /**
     * LIST 左推数据
     *
     * @param key   键
     * @param value 对象
     * @return 影响条数
     */
    @Override
    public <V> Long listLeftPush(String key, V value) {
        byte[] valBytes = this.serialize(value);
        byte[] keyBytes = this.serializeForString(key);
        return this.redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.lPush(keyBytes, valBytes);
            }
        });
    }

    /**
     * LIST 左读数据 读一个删除一个
     *
     * @param key   键
     * @param clazz 对象
     * @return 反系列化对象
     */
    @Override
    public <v> v listLeftPop(String key, Class<v> clazz) {
        final byte[] keyBytes = this.serializeForString(key);
        byte[] hGetBytes = this.redisTemplate.execute(connection -> connection.lPop(keyBytes), true);
        return this.deserialize(hGetBytes, clazz);
    }

    /**
     * LIST 右推数据
     *
     * @param key   键
     * @param value 对象
     * @return 影响条数
     */
    @Override
    public <V> Long lRightPush(String key, V value) {
        byte[] valBytes = this.serialize(value);
        byte[] keyBytes = this.serializeForString(key);
        return this.redisTemplate.execute(new RedisCallback<Long>() {
            @Override
            public Long doInRedis(RedisConnection connection) throws DataAccessException {
                return connection.rPush(keyBytes, valBytes);
            }
        });
    }

    /**
     * LIST 右读数据
     *
     * @param key   键
     * @param clazz 对象
     * @return 反系列化对象
     */
    @Override
    public <V> V lRightPop(String key, Class<V> clazz) {
        final byte[] keyBytes = this.serializeForString(key);
        byte[] hGetBytes = this.redisTemplate.execute(connection -> connection.rPop(keyBytes), true);
        return this.deserialize(hGetBytes, clazz);
    }

    @Override
    public Long length(String key) {
        final byte[] keyBytes = this.serializeForString(key);
        return this.redisTemplate.execute(connection -> connection.lLen(keyBytes), true);
    }

    public <V> List<V> getList(String key, Class<V> clazz) {
        int begin = 0;
        int end = Integer.MAX_VALUE;
        return this.getList(key, begin, end, clazz);
    }

    /**
     * 指定查询区间
     *
     * @param key
     * @param begin
     * @param end
     * @param clazz
     * @return
     */
    @Override
    public <V> List<V> getList(String key, final int begin, final int end, final Class<V> clazz) {
        List<V> valueList = new ArrayList<>();
        try {
            final byte[] keyBytes = this.serializeForString(key);
            List<byte[]> valueBytes = this.redisTemplate.execute(connection -> connection.lRange(keyBytes, begin, end), true);
            if (CollectionUtils.isEmpty(valueBytes)) {
                return valueList;
            }
            Iterator<byte[]> iterator = valueBytes.iterator();
            while (iterator.hasNext()) {
                byte[] next = iterator.next();
                valueList.add(deserialize(next, clazz));
            }
        } catch (Exception var6) {
            log.error(var6.getMessage(), var6);
            return valueList;
        }
        return valueList;
    }

    /**
     * HASH反系列化
     *
     * @param entries
     * @param clazz
     * @return
     */
    private <V> Map<String, V> deserializeMap(Map<byte[], byte[]> entries, Class<V> clazz) {
        if (entries == null) {
            return new HashMap<>(0);
        } else {
            Map<String, V> map = new LinkedHashMap<>(entries.size());
            Iterator<?> var4 = entries.entrySet().iterator();

            while (var4.hasNext()) {
                Map.Entry<byte[], byte[]> entry = (Map.Entry) var4.next();
                map.put(entry.getKey() == null ? null : new String(entry.getKey()), this.deserialize(entry.getValue(), clazz));
            }

            return map;
        }
    }

    /**
     * 序列Map key和value
     *
     * @param originalMap
     * @return
     * @throws Exception
     */
    private <V> Map<byte[], byte[]> serializeMap(Map<String, V> originalMap) {
        Map<byte[], byte[]> serializedMap = new HashMap<>();

        try {
            for (Map.Entry<String, V> entry : originalMap.entrySet()) {
                byte[] keyBytes = entry.getKey().getBytes();
                byte[] valueBytes;

                if (entry.getValue() instanceof String) {
                    valueBytes = entry.getValue().toString().getBytes();
                } else if (entry.getValue() instanceof Integer) {
                    valueBytes = Integer.toString((Integer) entry.getValue()).getBytes();
                } else if (entry.getValue() instanceof Long) {
                    valueBytes = Long.toString((Long) entry.getValue()).getBytes();
                } else if (entry.getValue() instanceof Double) {
                    valueBytes = Double.toString((Double) entry.getValue()).getBytes();
                } else if (entry.getValue() instanceof Float) {
                    valueBytes = Float.toString((Float) entry.getValue()).getBytes();
                } else {
                    if (entry.getValue() != null) {
                        valueBytes = JSON.toJSONBytes(entry.getValue());
                    } else {
                        log.error("serializeMap -->Unsupported value type");
                        continue;
                    }
                }
                serializedMap.put(keyBytes, valueBytes);
            }
        } catch (Exception e) {
            log.error("serializeMap error ", e);
        }

        return serializedMap;
    }

    /**
     * 根据key查询hash类所有数据
     *
     * @param key
     * @param clazz
     * @return
     */
    @Override
    public <V> Map<String, V> hGetAll(String key, Class<V> clazz) {
        final byte[] keyBytes = this.serializeForString(key);
        if (this.hasKey(keyBytes)) {
            Map<byte[], byte[]> entries = this.redisTemplate.execute(connection -> connection.hGetAll(keyBytes), true);
            return this.deserializeMap(entries, clazz);
        } else {
            log.warn("没有找到【{}】的数据", key);
            return new HashMap<>();
        }
    }

    /**
     * 根据key-key精准查询hash类数据
     *
     * @param key   redis主key
     * @param field 副键
     * @param clazz 对象类型
     * @return
     */
    @Override
    public <V> V hGet(String key, String field, Class<V> clazz) {
        V hashData = null;
        try {
            final byte[] keyBytes = this.serializeForString(key);
            if (this.hasKey(keyBytes)) {
                final byte[] fileIdBytes = this.serializeForString(field);
                byte[] hGetBytes = this.redisTemplate.execute(connection -> connection.hGet(keyBytes, fileIdBytes), true);
                if (hGetBytes != null) {
                    hashData = deserialize(hGetBytes, clazz);
                }
            }
        } catch (Exception var6) {
            log.error(var6.getMessage(), var6);
        }
        return hashData;
    }

    @Override
    public <V> void hPut(String key, String field, V value) {
        try {
            final byte[] keyBytes = this.serializeForString(key);
            final byte[] hKeyBytes = this.serializeForString(field);
            final byte[] valBytes = this.serialize(value);
            this.redisTemplate.execute(new RedisCallback<Boolean>() {
                @Override
                public Boolean doInRedis(RedisConnection connection) throws DataAccessException {
                    return connection.hSet(keyBytes, hKeyBytes, valBytes);
                }
            });
        } catch (Exception e) {
            log.error("往hash存放数据失败【{}】【{}】，原因", key, field, e);
        }
    }

    @Override
    public <V> void hPutAll(String key, Map<String, V> value) {
        try {
            final byte[] keyBytes = this.serializeForString(key);
            Map<byte[], byte[]> serializeMap = this.serializeMap(value);
            if (!serializeMap.isEmpty()) {
                this.redisTemplate.execute(new RedisCallback<Object>() {
                    @Override
                    public Object doInRedis(RedisConnection connection) throws DataAccessException {
                        connection.hMSet(keyBytes, serializeMap);
                        return null;
                    }
                });
            } else {
                log.warn("没有【{}】系列化后的数据，不执行存储", key);
            }
        } catch (Exception e) {
            log.error("往hash存放数据失败【{}】，原因", key, e);
        }
    }

    @Override
    public void hDel(String key, String hKey) {
        this.redisTemplate.opsForHash().delete(key, hKey);
    }

    /**
     * @param key key值
     * @return 是否存在
     * @Description: 是否存在当前key
     * @Author: chenzumao
     */
    public boolean hasKey(String key) {
        if (ObjectUtils.isEmpty(key)) {
            return false;
        }
        return Boolean.TRUE.equals(this.redisTemplate.hasKey(key));
    }

    /**
     * @param keyBytes key值
     * @return 是否存在
     * @Description: 是否存在当前key
     * @Author: chenzumao
     */
    private boolean hasKey(final byte[] keyBytes) {
        if (null == keyBytes || keyBytes.length == 0) {
            return false;
        }
        return Boolean.TRUE.equals(this.redisTemplate.execute(connection -> connection.exists(keyBytes), true));
    }

    public Long delKey(String key) {
        log.debug("cache del key:{}", key);
        byte[] bytes = serializeForString(key);
        if (!hasKey(bytes)) {
            log.info("没有找到指定的key【{}】", key);
            return 0L;
        }
        return this.redisTemplate.execute(connection ->
                connection.del(bytes), true);
    }
}